package com.weiron;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author: weiRon
 * @description Flink批处理文件，DateStream方式
 * @date: 2023/4/11 16:54
 */
public class StreamWordsCount {
    public static void main(String[] args) {
        //跟DataSet大致一样的操作
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> source = env.readTextFile("input/words.txt");

        SingleOutputStreamOperator<Tuple2<String, Long>> operator = source.flatMap((String line, Collector<Tuple2<String, Long>> out) ->
        {// 将一行文本进行分词
            String[] words = line.split(" ");
            // 将每个单词转换成二元组输出
            for (String word : words) {
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));
        //按照 word 进行分组
        KeyedStream<Tuple2<String, Long>, String> keyedStream = operator.keyBy(d -> d.f0);

        //分组内进行聚合统计
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyedStream.sum(1);

        sum.print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
